package com.linkhouse.consumer;

import com.aliyun.dysmsapi20170525.models.SendSmsRequest;
import com.linkhouse.config.RedisConfig;
import com.linkhouse.demo.Demo;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Component
@RocketMQMessageListener(topic = "PhoneCode_Topic" , consumerGroup = "consumer_PhoneCode")
public class SmsConsumer implements RocketMQListener, RocketMQPushConsumerLifecycleListener {
    @Resource
    private RedisTemplate redisTemplate;

    @Override
    public void onMessage(Object o) {

    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt messageExt : list) {
                    try {
                        System.out.println("重试次数: " + messageExt.getReconsumeTimes());
                        System.out.println("重试消费主题: " + messageExt.getTopic());
                        String phone = new String(messageExt.getBody(),"utf-8");
                        System.out.println("消费内容: " + phone);

                        // 生成验证码
                        String code = RandomStringUtils.randomNumeric(6);
                        System.out.println("验证码为: " + code);

                        // 存入redis
                        RedisConfig redisConfig = new RedisConfig();
                        redisConfig.redisTemplate(redisTemplate.getConnectionFactory()).boundValueOps("CODE_"+phone).set(code);

                        // 设置有效时间
                        redisTemplate.expire(phone,10, TimeUnit.MINUTES);

                        // 调用方法
                        com.aliyun.dysmsapi20170525.Client client = Demo.createClient("LTAI4GCbdwnjCGwQj9Z3tr3N", "27ih4pc1W1tqey9X4mmCy1JUotAX7L");
                        SendSmsRequest sendSmsRequest = new SendSmsRequest()
                                .setPhoneNumbers(phone)
                                .setSignName("东科创想")
                                .setTemplateCode("SMS_162522027")
                                .setTemplateParam("{\"sms_code\":" + code + "}");
                        // 复制代码运行请自行打印 API 的返回值
                        client.sendSms(sendSmsRequest);
                        System.out.println("发送成功");
                    } catch (Exception e) {
                        e.printStackTrace();
                        System.out.println("出现异常: " + e.getMessage());
                        if(messageExt.getReconsumeTimes() >= 3) {
                            consumeConcurrentlyContext.setDelayLevelWhenNextConsume(-1);
                            // 重试次数达到三次 说明当前业务存在异常 无需继续重试下去 直接返回消费成功
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }
                        System.out.println("发送失败");
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }
}
